package com.main.crossdomaineventmanagement.event.core;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.main.crossdomaineventmanagement.event.exception.EventPublishException;
import com.main.crossdomaineventmanagement.event.model.EventMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class RocketMQTxEventPublisher implements EventPublisher {
    private final TransactionMQProducer producer;
    private final ObjectMapper objectMapper = new ObjectMapper()
            .registerModule(new JavaTimeModule());  // 仅注册 JavaTime 模块

    @Autowired
    public RocketMQTxEventPublisher(@Qualifier("transactionProducer") TransactionMQProducer producer) {
        this.producer = producer;
    }

    @Override
    public void publish(EventMessage event) {
        try {
            Message msg = new Message(
                    "TransactionEventsTopic",
                    event.getEventType(),
                    objectMapper.writeValueAsBytes(event)
            );
            msg.putUserProperty("source", event.getSource());
            SendResult result = producer.send(msg);
            log.info("事务消息发送成功: {}:{}", event.getEventType(), result.getMsgId());
            log.info(event.toString());
        }  catch (MQClientException e) {
            log.error("客户端配置错误: {}", e.getErrorMessage());
            throw new EventPublishException("配置错误", e);
        } catch (RemotingException e) {
            log.error("网络通信失败: {}", e.getMessage());
            throw new EventPublishException("网络异常", e);
        } catch (MQBrokerException e) {
            log.error("Broker 处理失败: {}", e.getErrorMessage());
            throw new EventPublishException("服务端错误", e);
        } catch (InterruptedException | JsonProcessingException e) {
            log.error("发送中断: {}", e.getMessage());
            Thread.currentThread().interrupt();
            throw new EventPublishException("发送中断", e);
        }
    }
}